package org.lemon.transport.mqtt.server.handler;

import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
import com.google.gson.JsonElement;
import com.google.gson.JsonNull;
import com.google.gson.JsonObject;
import com.google.gson.JsonSyntaxException;
import com.google.protobuf.InvalidProtocolBufferException;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import lombok.extern.slf4j.Slf4j;
import org.lemon.transport.context.adaptor.AdaptorException;
import org.lemon.transport.context.adaptor.JsonConverter;
import org.lemon.transport.context.adaptor.ProtoConverter;
import org.lemon.transport.context.auth.GetOrCreateDeviceFromGatewayResponse;
import org.lemon.transport.context.auth.TransportDeviceInfo;
import org.lemon.transport.context.service.DefaultTransportService;
import org.lemon.transport.context.service.TransportService;
import org.lemon.transport.context.service.TransportServiceCallback;
import org.lemon.transport.context.utils.StringUtils;
import org.lemon.transport.mqtt.server.adaptors.JsonMqttAdaptor;
import org.lemon.transport.mqtt.server.adaptors.ProtoMqttAdaptor;
import org.lemon.transport.mqtt.server.service.MqttTransportContext;
import org.lemon.transport.mqtt.server.session.DeviceSessionContext;
import org.lemon.transport.mqtt.server.session.GatewayDeviceSessionContext;
import org.lemon.transport.mqtt.server.model.MqttTopicMatcher;
import org.server.transport.gen.SessionProto;
import org.server.transport.gen.TransportApiProto;
import org.server.transport.gen.TransportProto;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ConcurrentReferenceHashMap;

import javax.annotation.Nullable;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import static org.lemon.transport.context.service.DefaultTransportService.*;

@Slf4j
public class GatewaySessionHandler {

    private static final String DEFAULT_DEVICE_TYPE = "default";
    private static final String CAN_T_PARSE_VALUE = "Can't parse value: ";
    private static final String DEVICE_NAME = "deviceName";

    private final MqttTransportContext context;
    private final TransportService transportService;
    private final TransportDeviceInfo gatewayDeviceInfo;
    private final UUID sessionId;
    private final DeviceSessionContext deviceSessionContext;
    private final ChannelHandlerContext channel;

    /** 设备信息创建锁 */
    private final ConcurrentMap<String, Lock> deviceCreationLockMap;
    private final ConcurrentMap<String, GatewayDeviceSessionContext> devices;
    private final ConcurrentMap<String, ListenableFuture<GatewayDeviceSessionContext>> deviceFutures;
    private final ConcurrentMap<MqttTopicMatcher, Integer> mqttQoSMap;

    public GatewaySessionHandler(DeviceSessionContext deviceSessionCtx, UUID sessionId) {
        this.context = deviceSessionCtx.getContext();
        this.sessionId = sessionId;
        this.deviceSessionContext = deviceSessionCtx;
        this.channel = deviceSessionCtx.getChannel();
        this.transportService = context.getTransportService();
        this.gatewayDeviceInfo = deviceSessionCtx.getDeviceInfo();

        this.deviceCreationLockMap = createWeakMap();
        this.devices = new ConcurrentHashMap<>();
        this.deviceFutures = new ConcurrentHashMap<>();
        this.mqttQoSMap = new ConcurrentHashMap<>();

    }

    ConcurrentReferenceHashMap<String, Lock> createWeakMap() {
        return new ConcurrentReferenceHashMap<>(16, ConcurrentReferenceHashMap.ReferenceType.WEAK);
    }

    public void onDeviceConnect(MqttPublishMessage mqttMsg) throws AdaptorException {

        switch (deviceSessionContext.getPayloadType()){
            case JSON:
                onDeviceConnectJson(mqttMsg);
                break;
            case PROTOBUF:
                onDeviceConnectProto(mqttMsg);
                break;
        }
    }

    public void onDeviceDisconnect(MqttPublishMessage mqttMsg) throws AdaptorException {

        switch (deviceSessionContext.getPayloadType()){
            case JSON:
                onDeviceDisconnectJson(mqttMsg);
                break;
            case PROTOBUF:
                onDeviceDisconnectProto(mqttMsg);
                break;
        }
    }

    public void onDeviceTelemetry(MqttPublishMessage mqttMsg) throws AdaptorException {

        int msgId = mqttMsg.variableHeader().packetId();
        ByteBuf payload = mqttMsg.payload();
        switch (deviceSessionContext.getPayloadType()){
            case JSON:
                onDeviceTelemetryJson(msgId, payload);
                break;
            case PROTOBUF:
                onDeviceTelemetryProto(msgId, payload);
                break;
        }
    }

    public void onDeviceClaim(MqttPublishMessage mqttMsg) throws AdaptorException {

        int msgId = mqttMsg.variableHeader().packetId();
        ByteBuf payload = mqttMsg.payload();
        switch (deviceSessionContext.getPayloadType()){
            case JSON:
                onDeviceClaimJson(msgId, payload);
                break;
            case PROTOBUF:
                onDeviceClaimProto(msgId, payload);
                break;
        }
    }

    public void onDeviceAttributes(MqttPublishMessage mqttMsg) throws AdaptorException {

        int msgId = mqttMsg.variableHeader().packetId();
        ByteBuf payload = mqttMsg.payload();
        switch (deviceSessionContext.getPayloadType()){
            case JSON:
                onDeviceAttributesJson(msgId, payload);
                break;
            case PROTOBUF:
                onDeviceAttributesProto(msgId, payload);
                break;
        }
    }

    public void onDeviceAttributesRequest(MqttPublishMessage mqttMsg) throws AdaptorException {
    }

    public void onDeviceRpc(MqttPublishMessage mqttMsg) throws AdaptorException {
    }

    private void onDeviceConnectJson(MqttPublishMessage mqttMsg) throws AdaptorException {
        JsonElement json = getJson(mqttMsg);
        String deviceName = checkDeviceName(getDeviceName(json));
        String deviceType = getDeviceType(json);
        processOnConnect(mqttMsg, deviceName, deviceType);
    }

    private void onDeviceConnectProto(MqttPublishMessage mqttMsg) throws AdaptorException {

        try {
            TransportApiProto.ConnectMsg connectProto = TransportApiProto.ConnectMsg.parseFrom(ProtoMqttAdaptor.toBytes(mqttMsg.payload()));
            String deviceName = checkDeviceName(connectProto.getDeviceName());
            String deviceType = StringUtils.isNullOrEmpty(connectProto.getDeviceType()) ? DEFAULT_DEVICE_TYPE : connectProto.getDeviceType();
            processOnConnect(mqttMsg, deviceName, deviceType);
        } catch (InvalidProtocolBufferException e) {
            throw new AdaptorException(e);
        }
    }

    private void processOnConnect(MqttPublishMessage msg, String deviceName, String deviceType) {

        log.trace("[{}] onDeviceConnect: {}", sessionId, deviceName);
        Futures.addCallback(onDeviceConnect(deviceName, deviceType), new FutureCallback<GatewayDeviceSessionContext>() {
            @Override
            public void onSuccess(@Nullable GatewayDeviceSessionContext result) {
                int msgId = msg.variableHeader().packetId();
                if (msgId > 0) {
                    channel.writeAndFlush(MqttTransportHandler.createMqttPubAckMsg(msgId));
                }
                log.trace("[{}] onDeviceConnectOk: {}", sessionId, deviceName);
            }

            @Override
            public void onFailure(Throwable t) {
                log.warn("[{}] Failed to process device connect command: {}", sessionId, deviceName, t);
            }
        }, context.getExecutor());
    }

    private void onDeviceClaimJson(int msgId, ByteBuf payload) throws AdaptorException {

        JsonElement json = JsonMqttAdaptor.validateJsonPayload(sessionId, payload);
        if (json.isJsonObject()){
            JsonObject jsonObject = json.getAsJsonObject();
            for (Map.Entry<String, JsonElement> deviceEntry : jsonObject.entrySet()) {
                String deviceName = deviceEntry.getKey();
                Futures.addCallback(checkDeviceConnected(deviceName),
                        new FutureCallback<GatewayDeviceSessionContext>() {
                            @Override
                            public void onSuccess(GatewayDeviceSessionContext deviceCtx) {
                                if (!deviceEntry.getValue().isJsonObject()) {
                                    throw new JsonSyntaxException(CAN_T_PARSE_VALUE + json);
                                }
                                try {
                                    Long deviceId = deviceSessionContext.getDeviceId();
                                    TransportProto.ClaimDeviceMsg claimDeviceMsg = JsonConverter.convertToClaimDeviceProto(deviceId, deviceEntry.getValue());
                                    processClaimDeviceMsg(deviceCtx, claimDeviceMsg, deviceName, msgId);
                                }catch (Throwable e){
                                    log.warn("[{}][{}] Failed to convert claim message: {}", gatewayDeviceInfo.getDeviceId(), deviceName, deviceEntry.getValue(), e);
                                }
                            }

                            @Override
                            public void onFailure(Throwable t) {
                                log.debug("[{}] Failed to process device claiming command: {}", sessionId, deviceName, t);
                            }
                        }, context.getExecutor());
            }
        }
    }

    private void onDeviceClaimProto(int msgId, ByteBuf payload) throws AdaptorException {

        try {

        }catch (Exception e){

        }
    }

    private void onDeviceAttributesJson(int msgId, ByteBuf payload) throws AdaptorException {

        JsonElement json = JsonMqttAdaptor.validateJsonPayload(sessionId, payload);
        if (json.isJsonObject()){
            JsonObject jsonObject = json.getAsJsonObject();
            for (Map.Entry<String, JsonElement> deviceEntry : jsonObject.entrySet()) {
                String deviceName = deviceEntry.getKey();
                Futures.addCallback(checkDeviceConnected(deviceName),
                        new FutureCallback<GatewayDeviceSessionContext>() {
                            @Override
                            public void onSuccess(@Nullable GatewayDeviceSessionContext deviceSessionContext) {
                                if (!deviceEntry.getValue().isJsonObject()){
                                    throw new JsonSyntaxException(CAN_T_PARSE_VALUE + deviceEntry.getValue());
                                }
                                TransportProto.PostAttributeMsg postAttributeMsg = JsonConverter
                                        .convertToAttributesProto(deviceEntry.getValue().getAsJsonObject());
                                assert deviceSessionContext != null;
                                processPostAttributesMsg(deviceSessionContext, postAttributeMsg, deviceName, msgId);
                            }

                            @Override
                            public void onFailure(Throwable t) {
                                log.debug("[{}] Failed to process device attributes command: {}", sessionId, deviceName, t);
                            }
                        }, context.getExecutor());
            }
        }
    }

    private void onDeviceAttributesProto(int msgId, ByteBuf payload) throws AdaptorException {

        try {
            byte[] payloadBytes = ProtoMqttAdaptor.toBytes(payload);
            TransportApiProto.GatewayAttributesMsg attributesMsgProto = TransportApiProto.GatewayAttributesMsg.parseFrom(payloadBytes);
            List<TransportApiProto.AttributesMsg> attributesMsgList = attributesMsgProto.getMsgList();
            if (CollectionUtils.isEmpty(attributesMsgList)){
                log.debug("[{}] Devices attributes keys list is empty for: [{}]", sessionId, gatewayDeviceInfo.getDeviceId());
                throw new IllegalArgumentException("[" + sessionId + "] Devices attributes keys list is empty for [" + gatewayDeviceInfo.getDeviceId() + "]");
            }
            attributesMsgList.forEach(attributesMsg -> {
                String deviceName = checkDeviceName(attributesMsg.getDeviceName());
                Futures.addCallback(checkDeviceConnected(deviceName),
                        new FutureCallback<GatewayDeviceSessionContext>() {
                            @Override
                            public void onSuccess(GatewayDeviceSessionContext deviceCtx) {
                                TransportProto.PostAttributeMsg attributeKvListProto = attributesMsg.getMsg();
                                if (attributeKvListProto == null){
                                    throw new IllegalArgumentException("Attributes list for device: " + deviceName + "is empty");
                                }
                                try {
                                    TransportProto.PostAttributeMsg postAttributeMsg = ProtoConverter.validatePostAttributesMsg(attributeKvListProto.toByteArray());
                                    processPostAttributesMsg(deviceCtx, postAttributeMsg, deviceName, msgId);
                                }catch (Throwable e){
                                    log.warn("[{}][{}] Failed to process device attributes command: {}", gatewayDeviceInfo.getDeviceId(), deviceName, attributeKvListProto, e);
                                }
                            }

                            @Override
                            public void onFailure(Throwable t) {

                            }
                        }, context.getExecutor());
            });
        }catch (Exception e){

        }
    }

    private void processPostAttributesMsg(GatewayDeviceSessionContext deviceCtx, TransportProto.PostAttributeMsg postAttributeMsg, String deviceName, int msgId) {
        transportService.process(deviceCtx.getSessionInfo(), postAttributeMsg, getPubAckCallback(channel, deviceName, msgId, postAttributeMsg));
    }

    public void deregisterSession(String deviceName) {

        GatewayDeviceSessionContext deviceSessionCtx = devices.remove(deviceName);
        if (deviceSessionCtx != null) {
            deregisterSession(deviceName, deviceSessionCtx);
        } else {
            log.debug("[{}] Device [{}] was already removed from the gateway session", sessionId, deviceName);
        }
    }

    public String getNodeId() {
        return context.getServiceId();
    }

    public UUID getSessionId() {
        return sessionId;
    }

    public int nextMsgId() {
        return deviceSessionContext.nextMsgId();
    }

    private void onDeviceDisconnectJson(MqttPublishMessage msg) throws AdaptorException {
        String deviceName = checkDeviceName(getDeviceName(getJson(msg)));
        processOnDisconnect(msg, deviceName);
    }

    private void onDeviceDisconnectProto(MqttPublishMessage msg) throws AdaptorException {

        try {
            TransportApiProto.ConnectMsg connectProto = TransportApiProto.ConnectMsg.parseFrom(ProtoMqttAdaptor.toBytes(msg.payload()));
            String deviceName = checkDeviceName(connectProto.getDeviceName());
            processOnDisconnect(msg, deviceName);
        } catch (InvalidProtocolBufferException e) {
            throw new AdaptorException(e);
        }
    }

    private void processOnDisconnect(MqttPublishMessage msg, String deviceName){

        deregisterSession(deviceName);
        int msgId = msg.variableHeader().packetId();
        if (msgId > 0) {
            channel.writeAndFlush(MqttTransportHandler.createMqttPubAckMsg(msgId));
        }
    }

    private ListenableFuture<GatewayDeviceSessionContext> onDeviceConnect(String deviceName, String deviceType) {
        GatewayDeviceSessionContext result = devices.get(deviceName);
        if (result != null){
            return Futures.immediateFuture(result);
        }
        Lock deviceCreationLock = deviceCreationLockMap.computeIfAbsent(deviceName, s -> new ReentrantLock());
        deviceCreationLock.lock();
        try {
            result = devices.get(deviceName);
            if (result == null) {
                return getDeviceCreationFuture(deviceName, deviceType);
            } else {
                return Futures.immediateFuture(result);
            }
        } finally {
            deviceCreationLock.unlock();
        }
    }

    private void onDeviceTelemetryJson(int msgId, ByteBuf payload) throws AdaptorException {

        JsonElement json = JsonMqttAdaptor.validateJsonPayload(sessionId, payload);
        if (json.isJsonObject()){
            JsonObject jsonObject = json.getAsJsonObject();
            for (Map.Entry<String, JsonElement> deviceEntry : jsonObject.entrySet()) {
                String deviceName = deviceEntry.getKey();
                Futures.addCallback(checkDeviceConnected(deviceName),
                        new FutureCallback<GatewayDeviceSessionContext>() {
                            @Override
                            public void onSuccess(GatewayDeviceSessionContext deviceSessionContext) {

                                if ( !deviceEntry.getValue().isJsonArray() ){
                                    throw new JsonSyntaxException(CAN_T_PARSE_VALUE + json);
                                }
                                try {
                                    TransportProto.PostTelemetryMsg postTelemetryMsg = JsonConverter
                                            .convertToTelemetryProto(deviceEntry.getValue().getAsJsonArray());
                                    processPostTelemetryMsg(deviceSessionContext, postTelemetryMsg, deviceName, msgId);
                                }catch (Exception e){
                                    log.warn("[{}][{}] Failed to convert telemetry: {}", gatewayDeviceInfo.getDeviceId(), deviceName, deviceEntry.getValue(), e);
                                    channel.close();
                                }
                            }

                            @Override
                            public void onFailure(Throwable t) {
                                log.debug("[{}] Failed to process device telemetry command: {}", sessionId, deviceName, t);
                            }
                        }, context.getExecutor());
            }
        }else {
            throw new JsonSyntaxException(CAN_T_PARSE_VALUE + json);
        }
    }

    private void onDeviceTelemetryProto(int msgId, ByteBuf payload) throws AdaptorException {

        try {
            TransportApiProto.GatewayTelemetryMsg telemetryMsgProto = TransportApiProto.GatewayTelemetryMsg.parseFrom(ProtoMqttAdaptor.toBytes(payload));
            List<TransportApiProto.TelemetryMsg> deviceMsgList = telemetryMsgProto.getMsgList();
            if (CollectionUtils.isEmpty(deviceMsgList)){
                log.debug("[{}] Devices telemetry messages is empty for: [{}]", sessionId, gatewayDeviceInfo.getDeviceId());
                throw new IllegalArgumentException("[" + sessionId + "] Devices telemetry messages is empty for [" + gatewayDeviceInfo.getDeviceId() + "]");
            }
            deviceMsgList.forEach(telemetryMsg ->{
                String deviceName = checkDeviceName(telemetryMsg.getDeviceName());
                Futures.addCallback(checkDeviceConnected(deviceName),
                        new FutureCallback<GatewayDeviceSessionContext>() {
                            @Override
                            public void onSuccess(GatewayDeviceSessionContext deviceSessionContext) {
                                TransportProto.PostTelemetryMsg msg = telemetryMsg.getMsg();
                                try {
                                    TransportProto.PostTelemetryMsg postTelemetryMsg = ProtoConverter.validatePostTelemetryMsg(msg.toByteArray());
                                    processPostTelemetryMsg(deviceSessionContext, postTelemetryMsg, deviceName, msgId);
                                }catch (Exception e){
                                    log.warn("[{}][{}] Failed to convert telemetry: {}", gatewayDeviceInfo.getDeviceId(), deviceName, msg, e);
                                    channel.close();
                                }
                            }

                            @Override
                            public void onFailure(Throwable t) {
                                log.debug("[{}] Failed to process device telemetry command: {}", sessionId, deviceName, t);
                            }
                        }, context.getExecutor());
            });
        } catch (InvalidProtocolBufferException e) {
            throw new AdaptorException(e);
        }
    }

    private void processPostTelemetryMsg(GatewayDeviceSessionContext deviceCtx, TransportProto.PostTelemetryMsg postTelemetryMsg, String deviceName, int msgId) {
        transportService.process(deviceCtx.getSessionInfo(), postTelemetryMsg, getPubAckCallback(channel, deviceName, msgId, postTelemetryMsg));
    }

    /**
     * 处理认领设备信息
     * @param deviceCtx
     * @param claimDeviceMsg
     * @param deviceName
     * @param msgId
     */
    private void processClaimDeviceMsg(GatewayDeviceSessionContext deviceCtx, TransportProto.ClaimDeviceMsg claimDeviceMsg, String deviceName, int msgId) {
        transportService.process(deviceCtx.getSessionInfo(), claimDeviceMsg, getPubAckCallback(channel, deviceName, msgId, claimDeviceMsg));
    }

    private ListenableFuture<GatewayDeviceSessionContext> getDeviceCreationFuture(String deviceName, String deviceType) {

        SettableFuture<GatewayDeviceSessionContext> futureSet = SettableFuture.create();
        ListenableFuture<GatewayDeviceSessionContext> future = deviceFutures.putIfAbsent(deviceName, futureSet);
        if (future != null){
            return future;
        }
        try {
            TransportApiProto.GetOrCreateDeviceFromGatewayRequestMsg requestMsg =
                    TransportApiProto.GetOrCreateDeviceFromGatewayRequestMsg.newBuilder()
                            .setDeviceName(deviceName)
                            .setDeviceType(deviceType)
                            .setGatewayId(gatewayDeviceInfo.getDeviceId())
                            .build();
            transportService.process(requestMsg, new TransportServiceCallback<>() {

                @Override
                public void onSuccess(GetOrCreateDeviceFromGatewayResponse msg) {
                    GatewayDeviceSessionContext deviceSessionCtx = new GatewayDeviceSessionContext(GatewaySessionHandler.this, msg.getDeviceInfo(), msg.getDeviceProfile(), mqttQoSMap, transportService);
                    if (devices.putIfAbsent(deviceName, deviceSessionCtx) == null){
                        log.trace("[{}] First got or created device [{}], type [{}] for the gateway session", sessionId, deviceName, deviceType);
                        SessionProto.SessionInfoProto sessionInfo = deviceSessionCtx.getSessionInfo();
                        transportService.registerAsyncSession(sessionInfo, deviceSessionCtx);
                        TransportProto.TransportToDeviceActorMsg toDeviceActorMsg = TransportProto.TransportToDeviceActorMsg.newBuilder()
                                .setSessionInfo(sessionInfo)
                                .setSessionEvent(SESSION_EVENT_MSG_OPEN)
                                .setSubscribeToAttributes(SUBSCRIBE_TO_ATTRIBUTE_UPDATES_ASYNC_MSG)
                                .setSubscribeToRPC(SUBSCRIBE_TO_RPC_ASYNC_MSG)
                                .build();
                        transportService.process(toDeviceActorMsg, null);
                    }
                    futureSet.set(devices.get(deviceName));
                    deviceFutures.remove(deviceName);
                }

                @Override
                public void onError(Throwable e) {
                    log.warn("[{}] Failed to process device connect command: {}", sessionId, deviceName, e);
                    futureSet.setException(e);
                    deviceFutures.remove(deviceName);
                }
            });
            return futureSet;
        }catch (Throwable e){
            deviceFutures.remove(deviceName);
            throw e;
        }
    }

    private ListenableFuture<GatewayDeviceSessionContext> checkDeviceConnected(String deviceName) {
        GatewayDeviceSessionContext ctx = devices.get(deviceName);
        if (ctx == null) {
            log.debug("[{}] Missing device [{}] for the gateway session", sessionId, deviceName);
            return onDeviceConnect(deviceName, DEFAULT_DEVICE_TYPE);
        } else {
            return Futures.immediateFuture(ctx);
        }
    }

    private void deregisterSession(String deviceName, GatewayDeviceSessionContext deviceSessionCtx) {
        transportService.deregisterSession(deviceSessionCtx.getSessionInfo());
        transportService.process(deviceSessionCtx.getSessionInfo(), DefaultTransportService.SESSION_EVENT_MSG_CLOSED, null);
        log.debug("[{}] Removed device [{}] from the gateway session", sessionId, deviceName);
    }

    private JsonElement getJson(MqttPublishMessage mqttMsg) throws AdaptorException {
        return JsonMqttAdaptor.validateJsonPayload(sessionId, mqttMsg.payload());
    }

    private String checkDeviceName(String deviceName) {
        if (StringUtils.isNullOrEmpty(deviceName)) {
            throw new RuntimeException("Device name is empty!");
        } else {
            return deviceName;
        }
    }

    private String getDeviceName(JsonElement json) {
        return json.getAsJsonObject().get(DEVICE_NAME).getAsString();
    }

    private String getDeviceType(JsonElement json) {
        JsonElement type = json.getAsJsonObject().get("deviceType");
        return type == null || type instanceof JsonNull ? DEFAULT_DEVICE_TYPE : type.getAsString();
    }

    private <T> TransportServiceCallback<Void> getPubAckCallback(final ChannelHandlerContext ctx, final String deviceName, final int msgId, final T msg) {
        return new TransportServiceCallback<Void>() {
            @Override
            public void onSuccess(Void dummy) {
                log.trace("[{}][{}] Published msg: {}", sessionId, deviceName, msg);
                if (msgId > 0) {
                    ctx.writeAndFlush(MqttTransportHandler.createMqttPubAckMsg(msgId));
                }
            }

            @Override
            public void onError(Throwable e) {
                log.trace("[{}] Failed to publish msg: {} for device: {}", sessionId, msg, deviceName, e);
                ctx.close();
            }
        };
    }
}
